#include "sys.h"
#include "log.h"
#include "dial.h"
#include "cJSON.h"
#include "tcp_tactics.h"
#include "kafka.h"

void* do_tcp_tactics(void* arg)
{
    tcp_tactics_t* tcp = (tcp_tactics_t*)arg;
    tcpTactics_t* ptr;
    struct tcpTactics_head* head = &tcp->head;
    cJSON *result, *obj, *arr;

    dial_arg_t dial_arg = { 0 };
    dial_result_t* dial_result = NULL;
    tcp_result_t* tcp_result = NULL;
    char tmp_str[32] = { 0 };
    char topic[32] = { 0 };
    char *post_data = NULL, reportTime[32], collectTime[32];
    if (tcp->count == 0)
        return NULL;
    result = cJSON_CreateObject();
    cJSON_AddStringToObject(result, "dialTaskId", tcp->taskId);
    cJSON_AddStringToObject(result, "version", tcp->version);
    cJSON_AddStringToObject(result, "indexType", "tcp");
    get_timestring(reportTime);
    cJSON_AddStringToObject(result, "reportTime", reportTime);
    get_timestring(collectTime);
    cJSON_AddStringToObject(result, "collectTime", collectTime);
    // pthread_mutex_lock(&tcp->mutex);
    arr = cJSON_CreateArray();
    TAILQ_FOREACH(ptr, head, entry)
    {
        obj = cJSON_CreateObject();
        dial_arg.in.tcp_arg.ip = wd_ipaddr(ptr->host);
        dial_arg.in.tcp_arg.port = htons(ptr->port);
        dial_arg.in.tcp_arg.timeout = (ptr->timeout);
        dial_result = (dial_result_t*)dial_tcp(&dial_arg);
        tcp_result_print(&dial_result->out.tcp_result);

        tcp_result = &dial_result->out.tcp_result;
        cJSON_AddNumberToObject(obj, "connected", tcp_result->connected);
        sprintf(tmp_str, "%.2f", tcp_result->latency);
        cJSON_AddStringToObject(obj, "delay", tmp_str);
        strcpy(topic, ptr->kafkaTopic);
        cJSON_AddStringToObject(obj, "host", ptr->host);
        cJSON_AddStringToObject(obj, "businessTag", ptr->businessTag);
        cJSON_AddItemToArray(arr, obj);
        free(dial_result);
    }
    cJSON_AddItemToObject(result, "tcpList", arr);
    post_data = cJSON_Print(result);
    log_debug("TCP post_data:%s", post_data);
    kafka_send(post_data, strlen(post_data), tcp->kafka_url, topic);
    free(post_data);
    cJSON_Delete(result);
    // pthread_cond_signal(&tcp->cond);
    // pthread_mutex_unlock(&tcp->mutex);
    return NULL;
}